package com.atguigu.edu.realtime.dws.app;
import com.atguigu.edu.realtime.common.base.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import com.atguigu.edu.realtime.common.util.HBaseUtil;
import org.apache.hadoop.hbase.client.Connection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.TradeClassStatistics;
import com.atguigu.edu.realtime.common.bean.TradePaperStatistics;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.DimAsyncFunction;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

public class DwsExamineClassStatistics extends BaseApp {
    public static void main(String[] args) {
        new DwsExamineClassStatistics().start(
                11223,
                4,
                "dws_exam_class_exam_statistics_window",
                "dwd_exam_exam"
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            collector.collect(jsonObj);
                        }
                    }
                }
        );


        SingleOutputStreamOperator<JSONObject> watermarkDS = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        return jsonObject.getLong("ts")*1000;
                                    }
                                }
                        )
        );
        KeyedStream<JSONObject, String> jsonObjKeyedDS = watermarkDS.keyBy(jsonObj -> jsonObj.getString("course_id"));
        //jsonObjKeyedDS.print();
        SingleOutputStreamOperator<TradeClassStatistics> beanDS = jsonObjKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, TradeClassStatistics>() {
                    MapState<String, Long> userCount;
                    Long userSum = 0L;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        MapStateDescriptor<String, Long> userCt = new MapStateDescriptor<>("userCount", String.class, Long.class);
                        userCount = getRuntimeContext().getMapState(userCt);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TradeClassStatistics>.Context context, Collector<TradeClassStatistics> collector) throws Exception {
                        String userId = jsonObject.getString("user_id");
                        if (userCount.get(userId) == null ) {
                            userCount.put(userId, 1L);
                            userSum = 1L;
                        }
                        String courseId = jsonObject.getString("course_id");
                        Long durationSec = jsonObject.getLong("duration_sec");
                        Double source = jsonObject.getDouble("score");
                        Long ts = jsonObject.getLong("ts");
                        collector.collect(new TradeClassStatistics(
                                "", "", "",
                                courseId, "", userSum, source, durationSec, ts
                        ));
                    }
                }
        );

        AllWindowedStream<TradeClassStatistics, TimeWindow> windowAllDS = beanDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<TradeClassStatistics> reduceDS = windowAllDS.reduce(
                new ReduceFunction<TradeClassStatistics>() {
                    @Override
                    public TradeClassStatistics reduce(TradeClassStatistics v1, TradeClassStatistics v2) throws Exception {
                        v1.setTestExamPeople(v1.getTestExamPeople() + v2.getTestExamPeople());
                        v1.setTestAvgScore(v1.getTestAvgScore() + v2.getTestAvgScore());
                        v1.setTestAvgTime(v1.getTestAvgTime() + v2.getTestAvgTime());
                        return v1;
                    }
                },
                new AllWindowFunction<TradeClassStatistics, TradeClassStatistics, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<TradeClassStatistics> iterable, Collector<TradeClassStatistics> collector) throws Exception {
                        TradeClassStatistics orderBean = iterable.iterator().next();
                        orderBean.setStt(DateFormatUtil.tsToDateTime(timeWindow.getStart()));
                        orderBean.setEdt(DateFormatUtil.tsToDateTime(timeWindow.getEnd()));
                        orderBean.setCurDate(DateFormatUtil.tsToDate(timeWindow.getStart()));
                        collector.collect(orderBean);
                    }
                }
        );
        //reduceDS.print();
/*        SingleOutputStreamOperator<TradeClassStatistics> courseInfoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeClassStatistics>() {
                    @Override
                    protected void addDim(TradeClassStatistics obj, JSONObject dimJsonObj) {
                        obj.setCourse_name(dimJsonObj.getString("course_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_course_info";
                    }

                    @Override
                    public String getRowKey(TradeClassStatistics obj) {
                        return obj.getCourse_id();
                    }
                }, 10, TimeUnit.SECONDS
        );*/

        SingleOutputStreamOperator<TradeClassStatistics> realDS = reduceDS.map(
                new RichMapFunction<TradeClassStatistics, TradeClassStatistics>() {
                    Connection hBaseconn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hBaseconn = HBaseUtil.getHBaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hBaseconn);
                    }

                    @Override
                    public TradeClassStatistics map(TradeClassStatistics tradeClassStatistics) throws Exception {
                        String courseId = tradeClassStatistics.getCourse_id();
                        JSONObject dimBaseCourseInfo = HBaseUtil.getRow(hBaseconn, "edu", "dim_course_info", courseId, JSONObject.class);
                        tradeClassStatistics.setCourse_name(dimBaseCourseInfo.getString("course_name"));
                        return tradeClassStatistics;
                    }
                }
        );
        realDS.print();
        realDS.map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_exam_class_exam_statistics_window"));

    }
}
